package opdb

import (
	"database/sql/driver"
	"errors"
	"fmt"
	"log"
	"time"

	opio "gitee.com/haodreams/godriver/opdb/opio/driver"
)

// EventRows is an iterator over an executed query's results.
type EventRows struct {
	valsChan  chan []driver.Value
	columns   []string
	over      chan bool
	numColumn int
	err       error
	sub       *opio.Subscribe
}

func newEventRows(sub *opio.Subscribe, args []driver.Value, maxEventSize int) (e *EventRows, err error) {
	if len(args) == 0 {
		return nil, errors.New("必须配置需要订阅的测点")
	}

	keyName := "GN"
	var tags interface{}

	if _, ok := args[0].(string); ok {
		names := make([]string, 0, len(args))
		for i := range args {
			if s, ok := args[i].(string); ok {
				names = append(names, s)
			}
		}
		tags = names
	} else {
		keyName = "ID"
		ids := make([]string, 0, len(args))
		for i := range args {
			if s, ok := args[i].(string); ok {
				ids = append(ids, s)
			}
		}
		tags = ids
	}

	e = new(EventRows)
	e.sub = sub
	e.valsChan = make(chan []driver.Value, maxEventSize)
	e.over = make(chan bool)
	err = sub.InitSubscribe(tags, keyName, func(res *opio.Response) {
		errno := res.GetErrNo()
		if errno == 0 {
			e.dowithDataSet(res.GetDataSet())
			return
		}
		e.err = fmt.Errorf("error code:%d", errno)
		e.valsChan <- nil
		log.Println("subscribe:", e.err)
	})
	<-e.over
	return
}

func (m *EventRows) dowithDataSet(ds *opio.OPDataSet) {
	defer func() {
		if e := recover(); e != nil {
			m.err = errors.New(fmt.Sprint(e))
		}
	}()
	if len(m.columns) == 0 {
		cols := ds.GetColumns()
		m.columns = make([]string, len(cols))
		m.numColumn = len(m.columns)
		for i, col := range cols {
			m.columns[i] = col.GetName()
		}
		m.over <- true
	}
	n := len(m.columns)
	for ok, err := ds.Next(); ok && err == nil; ok, err = ds.Next() {
		vals := make([]driver.Value, n)
		for i := 0; i < n; i++ {
			dest, err := ds.GetValue(uint32(i))
			if err != nil {
				m.err = err
				m.valsChan <- nil
				return
			}

			if t, ok := dest.(time.Time); ok {
				dest = t.Unix()
			} else if t, ok := dest.(bool); ok {
				if t {
					dest = 1
				} else {
					dest = 0
				}
			}

			vals[i] = dest
		}
		m.valsChan <- vals
	}
}

// Columns returns the names of the columns. The number of
// columns of the result is inferred from the length of the
// slice. If a particular column name isn't known, an empty
// string should be returned for that entry.
func (m *EventRows) Columns() []string {
	return m.columns
}

// Close closes the rows iterator.
func (m *EventRows) Close() error {
	if m.sub != nil {
		m.sub.Close()
		m.sub = nil
		fmt.Println("realtime 关闭订阅数据")
	}
	return nil
}

// Next is called to populate the next row of data into
// the provided slice. The provided slice will be the same
// size as the Columns() are wide.
//
// Next should return io.EOF when there are no more rows.
//
// The dest should not be written to outside of Next. Care
// should be taken when closing Rows not to modify
// a buffer held in dest.
func (m *EventRows) Next(dest []driver.Value) error {
	if m.err != nil {
		return m.err
	}
	vals := <-m.valsChan
	if m.err != nil {
		return m.err
	}
	n := len(vals)
	for i := range dest {
		if i >= n {
			return errors.New("列数不一致")
		}
		dest[i] = vals[i]
	}
	return m.err
}
